package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator
import org.apache.spark.{SparkConf, SparkContext}

object Code34PageRank1 {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("LogDataCompute")
    val sc = new SparkContext(conf)

    val pageReleaseRDD: RDD[(String, List[String])] = sc
      .textFile("spark_code/data/page.txt")
      .map {
        case line =>
          val splitRes: Array[String] = line.split("->")
          val page: String = splitRes(0)
          val outPageList = splitRes(1).split(",").toList
          (page, outPageList)
      }

    pageReleaseRDD.cache()

    var pageListPr: RDD[(String, List[String], Double)] = pageReleaseRDD
      // 1.整理数据 对 A->B,D 按照 -> 切分页面，同时对指向的多个页面转换成数组或List
      .map {
        case (page, outPageList) =>
          (page, outPageList, 1.0)
      }


    var whileNum = 10
    while (whileNum > 0) {
      pageListPr =
        pageListPr
          // 2.第一轮计算 对页面进行打分
          .flatMap {
            case (page, outPageList, pr) => {
              val avgPr = pr / outPageList.size.toDouble
              outPageList.map {
                case page => (page, avgPr)
              }
            }
          }
          // 3.对每个页面统计获得的分数
          .groupBy(_._1)
          .mapValues(_.map(_._2).sum)
          .join(pageReleaseRDD)
          .map {
            case (page, (pr, outPageList)) => {
              println((page, outPageList, pr))
              (page, outPageList, pr)
            }
          }


      whileNum -= 1
    }

    pageListPr
      .foreach(println)

    while (true) {

    }
  }
}
